package com.shujia.flink.soure

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

import com.shujia.flink.soure.Demo3ReadMysql.Student
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.functions.source.{ParallelSourceFunction, RichParallelSourceFunction, RichSourceFunction, SourceFunction}

object Demo3ReadMysql {

  def main(args: Array[String]): Unit = {

    //创建flink的运行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    env.setParallelism(2)


    //使用自定义source
    val mysqlDS: DataStream[Student] = env.addSource(new MysqlSourceFunction)


    //统计班级人数
    val countDS: DataStream[(String, Int)] = mysqlDS.map(stu => (stu.clazz, 1)).keyBy(_._1).sum(1)


    /**
      * 在 flink代码中可以没有sink 前面的代码也会执行
      *
      */
    countDS.print()


    env.execute()

  }

  case class Student(id: Int, name: String, age: Int, gender: String, clazz: String)

}


/**
  * SourceFunction : 单线程读取数据
  * RichSourceFunction ：  单线程读取数据，在SourceFunction的基础上多个open和close方法
  * ParallelSourceFunction: 并行的source
  * RichParallelSourceFunction： 并行的source
  *
  */


class MysqlSourceFunction extends RichSourceFunction[Student] {

  /**
    * open方法在tun 方法之前执行，一般用户打开连接
    *
    */

  var con: Connection = _

  override def open(parameters: Configuration): Unit = {
    println("open")
    //1、加载驱动
    Class.forName("com.mysql.jdbc.Driver")

    //创建连接
    con = DriverManager.getConnection("jdbc:mysql://master:3306/student", "root", "123456")

  }


  /**
    * close 在run方法之后执行，一般用户回收资源
    *
    */

  override def close(): Unit = {
    println("close")
    con.close()
  }

  override def run(ctx: SourceFunction.SourceContext[Student]): Unit = {

    println("run方法开始运行")


    //查询数据
    val stat: PreparedStatement = con.prepareStatement("select * from student")

    val resultSet: ResultSet = stat.executeQuery()


    //解析数据
    while (resultSet.next()) {
      val id: Int = resultSet.getInt("id")
      val name: String = resultSet.getString("name")
      val age: Int = resultSet.getInt("age")
      val gender: String = resultSet.getString("gender")
      val clazz: String = resultSet.getString("clazz")

      println("数据发送到下游")


      //将数据发送到下游
      ctx.collect(Student(id, name, age, gender, clazz))
    }
  }

  override def cancel(): Unit = {

  }
}
